package com.snopy.mq.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

/**
 * @author snopy
 * @version 1.0.0
 * @classname KConsumer
 * @description
 * @email 77912204@qq.com
 * @date 2022/06/08 14:30
 */
public class KConsumer {
    public static void main(String[] args) {
        //1.配置消费者启动参数
        Properties props = new Properties();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.164.101:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());


        //允许自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);
        //自动提交时间
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);
        props.put(ConsumerConfig.GROUP_ID_CONFIG,"start-test");
        //2.传递kafka启动参数，创建消费者
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(props);
        //3.订阅topic
        consumer.subscribe(Collections.singleton("test"));
        //4.接受消息
        while(true){
            ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(100));
            records.forEach(record->{
                System.out.println("offset = "+record.offset()+",key = "+record.key()+", value = "+
                        record.value());
            });
        }
    }
}
